package com.cryann.mqttserver.service.Impl;

import com.cryann.mqttserver.modle.Accessor;
import com.cryann.mqttserver.modle.MqttSendMessage;
import com.cryann.mqttserver.modle.Topic;
import com.cryann.mqttserver.repository.AccessorRepository;
import com.cryann.mqttserver.repository.TopicRepository;
import com.cryann.mqttserver.service.MqttAccessor;
import com.cryann.mqttserver.service.MqttSubject;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Arrays;
import java.util.List;

/**
 * @author cRyann
 * @date 2018/12/20 0020 10:25
 * @description:
 */
@Slf4j
@Service("mqttSubject")
@Transactional
public class MqttSubjectImpl implements MqttSubject {

    @Autowired
    AccessorRepository accessorRepository;
    @Autowired
    TopicRepository topicRepository;
    @Autowired
    MqttAccessor mqttAccessor;

    @Override
    public boolean addSubject(Channel ctx, MqttSubscribeMessage msg) {
        List<MqttTopicSubscription> list = msg.payload().topicSubscriptions();
        list.stream().forEach(topic -> {
            Topic t = topicRepository.findByTopicName(topic.topicName()).orElse(new Topic());
            t.setTopicName(topic.topicName());
            Accessor accessor = mqttAccessor.subTopic(mqttAccessor.getAccessor(ctx), topic.topicName());
            t.getAccessors().add(accessor);
            topicRepository.saveAndFlush(t);
        });
        return true;
    }

    @Override
    public boolean delSubject(Channel ctx, MqttUnsubscribeMessage msg) {
        List<String> list = msg.payload().topics();
        log.info(Arrays.toString(list.toArray()));
        list.stream().forEach(topic -> {
            Topic t = topicRepository.findByTopicName(topic).orElse(new Topic());
            Accessor accessor = accessorRepository.findByChannelId(ctx.id().asShortText());
            accessor.getTopicName().remove(topic);
            accessorRepository.saveAndFlush(accessor);
            t.getAccessors().remove(accessor);
            if ( t.getAccessors().size() < 1 ) {
                topicRepository.delete(t);
            }else {
                topicRepository.saveAndFlush(t);
            }
        });
        return true;
    }

    @Override
    public void publicSub(Object mqttMessage) {
        mqttAccessor.sendMsg((MqttSendMessage) mqttMessage);
    }

    @Override
    public boolean pubAck(Channel ctx, MqttPublishMessage message) {
        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBACK, false, MqttQoS.AT_LEAST_ONCE, false, 2);
        MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(message.variableHeader().packetId());
        MqttMessage mqttMessage = MqttMessageFactory.newMessage(fixedHeader, variableHeader, null);
        ctx.writeAndFlush(mqttMessage);
        return true;
    }

    @Override
    public boolean pubrec(Channel ctx, MqttPublishMessage message) {

        MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBREC, false, MqttQoS.AT_MOST_ONCE, false, 2);
        MqttMessageIdVariableHeader variableHeader = MqttMessageIdVariableHeader.from(message.variableHeader().packetId());
        MqttMessage msg = MqttMessageFactory.newMessage(fixedHeader, variableHeader, null);
        log.info("SENT:[{}]", msg);
        ctx.writeAndFlush(msg);
        return false;
    }

}
